package com.mlog.njyjy.ndop.share.kafka.admin.config;

import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>
 * .
 * </p>
 *
 * @author <a href="mailto:yingxiufeng@mlogcn.com">alex</a>
 * @version v1.0, 2020/7/7
 */
@Configuration
@EnableConfigurationProperties({KafkaAdminProperties.class})
public class kafkaAdminConfiguration {

    private final KafkaAdminProperties properties;

    public kafkaAdminConfiguration(KafkaAdminProperties properties) {
        this.properties = properties;
    }

    @Bean
    public KafkaAdminClient kafkaAdminClient() {
        PropertyMapper map = PropertyMapper.get();
        Map<String, Object> adminProp = new HashMap<>(2);
        map.from(this.properties::getBootstrapServers).when(servers -> servers.size() > 0)
                .to(servers -> adminProp.put("bootstrap.servers", servers.stream().collect(Collectors.joining(","))));
        map.from(this.properties::getClientId).whenHasText().to(str -> adminProp.put("client.id", str));

        map.from(this.properties::getRequestTimeout).whenNonNull()
                .to(duration -> adminProp.put("request.timeout.ms", duration.toMillis()));

        map.from(this.properties::getRetries).when(val -> val > 0)
                .to(val -> adminProp.put("retries", val));

        // oauth authenticate
//        adminProp.put();

        // create admin client
        return (KafkaAdminClient) KafkaAdminClient.create(adminProp);
    }
}
